package streamproc.filtercomponents;

import java.util.HashMap;
import java.util.Iterator;
import java.util.Vector;
import event.Event;
import event.EventFormatManager;
import event.DEFINES.CEPLEventDataFieldType;
import event.UnknownDataFieldException;
import streamdef.StreamCondition;
import streamdef.StreamDefinitionManager;
import streamdef.ConditionStatement;
import streamdef.ConditionStatementFieldFromOneStream;
import streamdef.ConditionStatementFieldFromTwoStreams;
import streamdef.ConditionStatementTimeRelationTwoStreams;
import streamdef.SubCondition;
import streamdef.DEFINES.FieldRelationType;
import streamdef.DEFINES.ConditionStatementType;
import streamproc.ChainComponent;

public class EventFilterComponent extends ChainComponent {
	private static EventFilterComponent sInstance = null;

	public static EventFilterComponent instance() {
		if (sInstance == null)
			sInstance = new EventFilterComponent();

		return sInstance;
	}

	private HashMap<Long, HashMap<String, Vector<FilterStatement>>> streamIDToFilterStatements = null;

	// <fs> has a valid _external_ streamID and field
	private FilterStatement convertToRequiredValueType(FilterStatement fs) {
		fs.valueType = EventFormatManager.instance().getFieldValueType(
				fs.streamID, fs.field);

		switch (fs.valueType) {
		case CHAR:
			fs.charValue = (char) fs.doubleValue;
			fs.charValue2 = (char) fs.doubleValue2;
			break;
		case SHORT:
			fs.shortValue = (short) fs.doubleValue;
			fs.shortValue2 = (short) fs.doubleValue2;
			break;
		case INT:
			fs.intValue = (int) fs.doubleValue;
			fs.intValue2 = (int) fs.doubleValue2;
			break;
		case LONG:
			fs.longValue = (long) fs.doubleValue;
			fs.longValue2 = (long) fs.doubleValue2;
			break;
		case FLOAT:
			fs.floatValue = (float) fs.doubleValue;
			fs.floatValue2 = (float) fs.doubleValue2;
			break;
		case DOUBLE:
			fs.doubleValue = (double) fs.doubleValue;
			fs.doubleValue2 = (double) fs.doubleValue2;
			break;
		case TEXT:
			fs.stringValue = fs.stringValue;
			break;
		}

		return fs;
	}

	private FilterStatement convertQueryStatement(
			ConditionStatementFieldFromOneStream qs, long extID, long intID,
			String field, boolean isAND) {
		FilterStatement newFS = new FilterStatement();
		newFS.field = field;
		newFS.streamID = intID;
		newFS.relationType = qs.getRelation();
		newFS.valueType = EventFormatManager.instance().getFieldValueType(
				extID, field);

		switch (newFS.valueType) {
		case CHAR:
			newFS.charValue = (Character) qs.getFieldValue();
			newFS.doubleValue = newFS.charValue;
			break;
		case SHORT:
			newFS.shortValue = (Short) qs.getFieldValue();
			newFS.doubleValue = newFS.shortValue;
			break;
		case INT:
			newFS.intValue = (Integer) qs.getFieldValue();
			newFS.doubleValue = newFS.intValue;
			break;
		case LONG:
			newFS.longValue = (Long) qs.getFieldValue();
			newFS.doubleValue = newFS.longValue;
			break;
		case FLOAT:
			newFS.floatValue = (Float) qs.getFieldValue();
			newFS.doubleValue = newFS.floatValue;
			break;
		case DOUBLE:
			newFS.doubleValue = (Double) qs.getFieldValue();
			newFS.doubleValue = newFS.doubleValue;
			break;
		case TEXT:
			newFS.stringValue = (String) qs.getFieldValue();
			break;
		}

		newFS.wasInOr = !isAND;
		return newFS;
	}

	private EventFilterComponent() {
		streamIDToFilterStatements = new HashMap<Long, HashMap<String, Vector<FilterStatement>>>();
	}

	public void reset() {
		if (streamIDToFilterStatements != null)
			streamIDToFilterStatements.clear();
	}

	protected void finalize() {
		reset();
	}

	public void init() {
		reset();

		Vector<Long> mainQueryIDs = StreamDefinitionManager.instance()
				.getStreamIDs();
		HashMap<Long, HashMap<String, Vector<FilterStatement>>> filtStatsForAllQueries = new HashMap<Long, HashMap<String, Vector<FilterStatement>>>();

		for (int i = 0; i < mainQueryIDs.size(); ++i) {
			long mqID = mainQueryIDs.get(i);
			HashMap<Long, HashMap<String, Vector<FilterStatement>>> filtStatsForMainQuery = getFilterStatementsForMainQuery(mqID);
			filtStatsForAllQueries = mergeFiltStatsMainQuery(
					filtStatsForAllQueries, filtStatsForMainQuery);
		}

		// fields with empty vectors of FilterStatements
		// -> wildcards -> just drop the fields out of the hashmap
		streamIDToFilterStatements = simplifyAndOrderResultingHashMap(filtStatsForAllQueries);
	}

	// Long = external ID
	private HashMap<Long, HashMap<String, Vector<FilterStatement>>> getFilterStatementsForMainQuery(
			long mqID) {
		HashMap<Long, HashMap<String, Vector<FilterStatement>>> resultExt = new HashMap<Long, HashMap<String, Vector<FilterStatement>>>();
		HashMap<Long, HashMap<String, Vector<FilterStatement>>> resultInt = new HashMap<Long, HashMap<String, Vector<FilterStatement>>>();

		StreamCondition mq = StreamDefinitionManager.instance().getParsedQuery(
				mqID);
		Vector<SubCondition> subConditions = mq.getSubQueries();

		for (int i = 0; i < subConditions.size(); ++i) {
			SubCondition sq = subConditions.get(i);
			HashMap<Long, HashMap<String, Vector<FilterStatement>>> filtStatsForSubQuery = getFilterStatementsForSubQuery(
					sq, mq);
			resultInt = addFiltStatsSubQuery(resultInt, filtStatsForSubQuery);
		}

		// merge the subQueries with AND
		resultInt = mergeFilterStatementsForAllIDsAndAllFields(resultInt, true);

		// convert internalID's to externalID's
		Iterator<Long> iteratorID = resultInt.keySet().iterator();
		while (iteratorID.hasNext()) {
			Long intID = (Long) iteratorID.next(); // internalID
			HashMap<String, Vector<FilterStatement>> intIDInMainQuery = resultInt
					.get(intID);
			long extID = mq.getExternalStreamID(intID);
			HashMap<String, Vector<FilterStatement>> extIDInMainQuery = resultExt
					.get(extID);

			if (extIDInMainQuery == null) {
				// new external ID for this query
				resultExt.put(extID, intIDInMainQuery);
			}

			Iterator<String> iteratorField = intIDInMainQuery.keySet()
					.iterator();
			while (iteratorField.hasNext()) {
				String field = (String) iteratorField.next();
				Vector<FilterStatement> intFieldsInMainQuery = intIDInMainQuery
						.get(field);
				for (int l = 0; l < intFieldsInMainQuery.size(); ++l)
					intFieldsInMainQuery.get(l).streamID = extID;

				if (extIDInMainQuery != null) {
					Vector<FilterStatement> extFieldsInMainQuery = extIDInMainQuery
							.get(field);

					if (extFieldsInMainQuery == null) {
						// new field for this external ID in this query
						extIDInMainQuery.put(field, intFieldsInMainQuery);
					} else {
						if (extFieldsInMainQuery.size() == 0) {
							// wildcard by another internalID
							// -> keep it
						} else {
							extFieldsInMainQuery.addAll(intFieldsInMainQuery);
						}
					}
				}
			}
		}

		// merge the externalID's with OR
		resultExt = mergeFilterStatementsForAllIDsAndAllFields(resultExt, false);

		HashMap<Long, HashMap<String, Vector<FilterStatement>>> result = new HashMap<Long, HashMap<String, Vector<FilterStatement>>>();
		Vector<Long> wildcardIDs = getWildCardsForMainQuery(mq); // externalIDs
		for (int i = 0; i < wildcardIDs.size(); ++i) {
			result.put(wildcardIDs.get(i),
					new HashMap<String, Vector<FilterStatement>>());
		}

		// drop the IDs for which a wildcard exists
		Iterator<Long> iteratorID_ = resultExt.keySet().iterator();
		while (iteratorID_.hasNext()) {
			Long ID = iteratorID_.next();
			if (!result.containsKey(ID))
				result.put(ID, resultExt.get(ID));
		}

		return result;
	}

	// Long = internal ID
	private HashMap<Long, HashMap<String, Vector<FilterStatement>>> getFilterStatementsForSubQuery(
			SubCondition sq, StreamCondition mq) {
		HashMap<Long, HashMap<String, Vector<FilterStatement>>> result = new HashMap<Long, HashMap<String, Vector<FilterStatement>>>();
		boolean isAND = sq.isANDStatement();
		Vector<ConditionStatement> conditionStatements = sq
				.getQueryStatements();

		for (int i = 0; i < conditionStatements.size(); ++i) {
			ConditionStatement qs = conditionStatements.get(i);

			if (qs.getType() == ConditionStatementType.FIELD_ONESTREAM) {
				ConditionStatementFieldFromOneStream qs_one_stream = (ConditionStatementFieldFromOneStream) qs;
				String field = qs_one_stream.getField();
				long internalID = qs_one_stream.getInternalStreamID();
				long externalID = mq.getExternalStreamID(internalID);

				// create filterstatement from predicate
				FilterStatement newFS = convertQueryStatement(qs_one_stream,
						externalID, internalID, field, isAND);

				// temporary hashmap, using _internalStreamID_
				// -> 2 predicates with same internalStreamID -> isAND
				HashMap<String, Vector<FilterStatement>> prevFilterStatements = result
						.get(internalID);

				if (prevFilterStatements == null) {
					// streamID not yet used
					prevFilterStatements = new HashMap<String, Vector<FilterStatement>>();
					Vector<FilterStatement> newV = new Vector<FilterStatement>();

					newV.add(newFS);
					prevFilterStatements.put(newFS.field, newV);
					result.put(newFS.streamID, prevFilterStatements);
				} else {
					Vector<FilterStatement> prevFS = prevFilterStatements
							.get(newFS.field);
					if (prevFS == null) {
						// field not yet used
						Vector<FilterStatement> newV = new Vector<FilterStatement>();
						newV.add(newFS);
						prevFilterStatements.put(newFS.field, newV);
					} else
						prevFS.add(newFS);
				}
			}
		}

		result = mergeFilterStatementsForAllIDsAndAllFields(result, isAND);

		return result;
	}

	// Long = externalIDs
	private Vector<Long> getWildCardsForMainQuery(StreamCondition mq) {
		// internal IDs that occur in BEFORE or JOIN
		Vector<Long> fromValidStatsMain = new Vector<Long>();
		// internal IDs that occur in equations
		Vector<Long> fromInvalidStatsMain = new Vector<Long>();

		Vector<SubCondition> subConditions = mq.getSubQueries();

		// each subquery
		for (int i = 0; i < subConditions.size(); ++i) {
			SubCondition sq = subConditions.get(i);
			Vector<ConditionStatement> queryStats = sq.getQueryStatements();
			boolean isAND = !sq.isANDStatement();

			// internal IDs that occur in BEFORE or JOIN -> wildcards
			Vector<Long> fromValidStatsSub = new Vector<Long>();
			// internal IDs that occur in equations -> no wildcards
			Vector<Long> fromInvalidStatsSub = new Vector<Long>();

			for (int j = 0; j < queryStats.size(); ++j) {
				ConditionStatement queryStat = queryStats.get(j);
				if (queryStat.getType() == ConditionStatementType.FIELD_ONESTREAM) {
					ConditionStatementFieldFromOneStream queryStatCast = (ConditionStatementFieldFromOneStream) queryStat;
					fromInvalidStatsSub.add(queryStatCast.internalStreamID);
				} else if (queryStat.getType() == ConditionStatementType.FIELD_TWOSTREAMS) {
					ConditionStatementFieldFromTwoStreams queryStatCast = (ConditionStatementFieldFromTwoStreams) queryStat;
					fromValidStatsSub.add(queryStatCast.internalStreamID1);
					fromValidStatsSub.add(queryStatCast.internalStreamID2);
				} else if (queryStat.getType() == ConditionStatementType.TIMERELATION_TWOSTREAMS) {
					ConditionStatementTimeRelationTwoStreams queryStatCast = (ConditionStatementTimeRelationTwoStreams) queryStat;
					fromValidStatsSub.add(queryStatCast.internalStreamID1);
					fromValidStatsSub.add(queryStatCast.internalStreamID2);
				}
			}

			// all possible wildcard IDs
			for (int j = 0; j < fromValidStatsSub.size(); ++j) {
				long internalID = fromValidStatsSub.get(j);

				if (fromInvalidStatsMain.contains(internalID)) {
					; // already not a wildcard -> leave it
				} else if (!isAND) { // subquery = OR
					if (!fromValidStatsMain.contains(internalID))
						fromValidStatsMain.add(internalID);
				} else if (isAND) { // subquery = AND
					if (fromInvalidStatsSub.contains(internalID)) {
						fromValidStatsMain.remove(new Long(internalID));
						fromInvalidStatsMain.add(internalID);
					} else {
						if (!fromValidStatsMain.contains(internalID))
							fromValidStatsMain.add(internalID);
					}
				}
			}

			for (int j = 0; j < fromInvalidStatsSub.size(); ++j) {
				long internalID = fromInvalidStatsSub.get(j);

				// not yet processed
				if (!fromValidStatsSub.contains(internalID)) {
					fromValidStatsMain.remove(new Long(internalID));
					fromInvalidStatsMain.add(internalID);
				}
			}
		}

		// if external ID's occur in valid and invalid statements
		// -> keep wildcard
		Vector<Long> fromValidStatsMainExt = new Vector<Long>();

		for (int j = 0; j < fromValidStatsMain.size(); ++j) {
			long extID = mq.getExternalStreamID(fromValidStatsMain.get(j));
			if (!fromValidStatsMainExt.contains(extID))
				fromValidStatsMainExt.add(extID);
		}

		return fromValidStatsMainExt;
	}

	// Long = externalIDs
	private HashMap<Long, HashMap<String, Vector<FilterStatement>>> mergeFiltStatsMainQuery(
			HashMap<Long, HashMap<String, Vector<FilterStatement>>> filtStatsForAllQueries,
			HashMap<Long, HashMap<String, Vector<FilterStatement>>> filtStatsForMainQuery) {
		HashMap<Long, HashMap<String, Vector<FilterStatement>>> result = filtStatsForAllQueries;

		// for each used ID in the new MainQuery -> merge its filterstatements
		// with the rest
		Iterator<Long> iteratorID = filtStatsForMainQuery.keySet().iterator();
		while (iteratorID.hasNext()) {
			long ID = (Long) iteratorID.next(); // ID

			HashMap<String, Vector<FilterStatement>> fieldToFilterStatements = filtStatsForMainQuery
					.get(ID);
			HashMap<String, Vector<FilterStatement>> fieldToFilterStatementsOri = filtStatsForAllQueries
					.get(ID);

			if (fieldToFilterStatements.size() == 0
					|| (fieldToFilterStatementsOri != null && fieldToFilterStatementsOri
							.size() == 0)) {
				// existing wildcard for this ID -> keep it
				result.put(ID, new HashMap<String, Vector<FilterStatement>>());
			} else if (fieldToFilterStatementsOri == null) {
				// ID not yet used in a query -> copy the filterStatements
				result.put(ID, fieldToFilterStatements);
			} else {
				// merge the two hashmaps, field by field
				HashMap<String, Vector<FilterStatement>> result_fieldToFilterStatements = new HashMap<String, Vector<FilterStatement>>();

				Iterator<String> iteratorField = fieldToFilterStatements
						.keySet().iterator();
				while (iteratorField.hasNext()) {
					String field = (String) iteratorField.next(); // field

					Vector<FilterStatement> filterStatements = fieldToFilterStatements
							.get(field);
					Vector<FilterStatement> filterStatementsOri = fieldToFilterStatementsOri
							.get(field);

					// the field was not yet used in another query
					// -> wildcard because of that
					if (filterStatementsOri == null) {
						result_fieldToFilterStatements.put(field,
								new Vector<FilterStatement>());
					}
					// existing wildcard for this field -> keep it by
					// adding an empty vector to the resulting hashmap
					else if (checkIfAnyIsWildcard(filterStatements)
							|| checkIfAnyIsWildcard(filterStatementsOri)) {
						result_fieldToFilterStatements.put(field,
								new Vector<FilterStatement>());
					} else {
						Vector<FilterStatement> result_filterStatements = filterStatements;
						if (filterStatementsOri != null)
							result_filterStatements.addAll(filterStatementsOri);

						// merge all the existing statements with OR
						result_filterStatements = mergeFilterStatementsForField(
								result_filterStatements, false);
						if (checkIfAnyIsWildcard(result_filterStatements))
							result_fieldToFilterStatements.put(field,
									new Vector<FilterStatement>());
						else
							result_fieldToFilterStatements.put(field,
									result_filterStatements);
					}

				}

				result.put(ID, result_fieldToFilterStatements);
			}
		}

		return result;
	}

	// Long = internalIDs
	private HashMap<Long, HashMap<String, Vector<FilterStatement>>> addFiltStatsSubQuery(
			HashMap<Long, HashMap<String, Vector<FilterStatement>>> filtStatsForMainQueryInt,
			HashMap<Long, HashMap<String, Vector<FilterStatement>>> filtStatsForSubQuery) {
		HashMap<Long, HashMap<String, Vector<FilterStatement>>> result = filtStatsForMainQueryInt;

		// add the subquery predicates to the main query
		Iterator<Long> iteratorID = filtStatsForSubQuery.keySet().iterator();
		while (iteratorID.hasNext()) {
			Long ID = (Long) iteratorID.next(); // ID
			HashMap<String, Vector<FilterStatement>> idInMainQuery = result
					.get(ID);
			HashMap<String, Vector<FilterStatement>> idInSubQuery = filtStatsForSubQuery
					.get(ID);

			if (idInMainQuery == null) {
				// not yet in mainquery
				result.put(ID, idInSubQuery);
			} else {
				Iterator<String> iteratorField = idInSubQuery.keySet()
						.iterator();
				while (iteratorField.hasNext()) {
					String field = (String) iteratorField.next();
					Vector<FilterStatement> fieldsInMainQuery = idInMainQuery
							.get(field);
					Vector<FilterStatement> fieldsInSubQuery = idInSubQuery
							.get(field);
					if (fieldsInMainQuery == null)
						idInMainQuery.put(field, fieldsInSubQuery);
					else {
						for (int l = 0; l < fieldsInSubQuery.size(); ++l) {
							fieldsInMainQuery.add(fieldsInSubQuery.get(l));
						}
					}
				}
			}
		}

		return result;
	}

	// Long = either
	private HashMap<Long, HashMap<String, Vector<FilterStatement>>> mergeFilterStatementsForAllIDsAndAllFields(
			HashMap<Long, HashMap<String, Vector<FilterStatement>>> prevFilterStatements,
			boolean isAND) {
		HashMap<Long, HashMap<String, Vector<FilterStatement>>> result = new HashMap<Long, HashMap<String, Vector<FilterStatement>>>();

		Iterator<Long> iteratorID = prevFilterStatements.keySet().iterator();
		while (iteratorID.hasNext()) {
			Long ID = (Long) iteratorID.next(); // ID
			HashMap<String, Vector<FilterStatement>> filterStatementsPerField = prevFilterStatements
					.get(ID);
			Iterator<String> iteratorField = filterStatementsPerField.keySet()
					.iterator();

			HashMap<String, Vector<FilterStatement>> newFilterStatementsPerField = new HashMap<String, Vector<FilterStatement>>();

			while (iteratorField.hasNext()) {
				String field = (String) iteratorField.next(); // field
				Vector<FilterStatement> filterStatements = filterStatementsPerField
						.get(field);

				boolean useAND = isAND;
				if (checkIfAnyWasInOr(filterStatements))
					useAND = false;

				Vector<FilterStatement> newFilterStatements = mergeFilterStatementsForField(
						filterStatements, useAND);

				newFilterStatementsPerField.put(field, newFilterStatements);
			}

			result.put(ID, newFilterStatementsPerField);
		}

		return result;
	}

	// if return.size() == 0 -> wildcard for this field
	private Vector<FilterStatement> mergeFilterStatementsForField(
			Vector<FilterStatement> prevFilterStatements, boolean isAND) {
		Vector<FilterStatement> temp = prevFilterStatements;
		Vector<FilterStatement> prevTemp = new Vector<FilterStatement>();

		while (temp.size() != prevTemp.size()) {
			if (temp.size() <= 1)
				return temp;
			else {
				prevTemp = temp;
				temp = mergeNewFilterStatementWithCurrentFilterStatement(
						prevTemp.get(0), prevTemp.get(1), isAND);
				for (int i = 2; i < prevTemp.size(); ++i) {
					for (int j = 0; j < temp.size(); ++j) {
						Vector<FilterStatement> newResults = mergeNewFilterStatementWithCurrentFilterStatement(
								prevTemp.get(i), temp.get(j), isAND);
						temp = filterOutDoubles(temp, newResults);
					}
				}
			}
		}

		boolean wildCardFound = false;
		for (int i = 0; i < temp.size() && !wildCardFound; ++i)
			if (temp.get(i).wildCard) {
				wildCardFound = true;
				temp.clear();
			}

		return temp;
	}

	private boolean checkIfAnyWasInOr(Vector<FilterStatement> v) {
		for (int i = 0; i < v.size(); ++i)
			if (v.get(i).wasInOr)
				return true;
		return false;

	}

	private boolean checkIfAnyIsWildcard(Vector<FilterStatement> v) {
		if (v == null)
			return false;

		if (v.size() == 0)
			return true;

		for (int i = 0; i < v.size(); ++i)
			if (v.get(i).wildCard)
				return true;
		return false;
	}

	// only add new predicates to <ori>
	private Vector<FilterStatement> filterOutDoubles(
			Vector<FilterStatement> ori, Vector<FilterStatement> newResults) {
		for (int i = 0; i < newResults.size(); ++i) {
			boolean found = false;
			FilterStatement fsNew = newResults.get(i);
			for (int j = 0; j < ori.size() && !found; ++j) {
				FilterStatement fsOri = ori.get(j);

				if (fsOri.field.compareToIgnoreCase(fsNew.field) == 0
						&& fsOri.streamID == fsNew.streamID
						&& fsNew.relationType == fsOri.relationType
						&& fsNew.valueType == fsOri.valueType) {
					if (fsNew.valueType == CEPLEventDataFieldType.TEXT) {
						found = fsNew.stringValue
								.compareToIgnoreCase(fsOri.stringValue) == 0;
					} else if (fsNew.relationType == FieldRelationType.EQUALS
							|| fsNew.relationType == FieldRelationType.EQUALSORGREATER
							|| fsNew.relationType == FieldRelationType.EQUALSORLESSER
							|| fsNew.relationType == FieldRelationType.GREATER
							|| fsNew.relationType == FieldRelationType.LESSER
							|| fsNew.relationType == FieldRelationType.NOTEQUALS) {
						found = fsNew.doubleValue == fsOri.doubleValue;
					} else if (fsNew.relationType == FieldRelationType.BETWEENEXCLUSIVE
							|| fsNew.relationType == FieldRelationType.BETWEENINCLUSIVELEFT
							|| fsNew.relationType == FieldRelationType.BETWEENINCLUSIVELEFTANDRIGHT
							|| fsNew.relationType == FieldRelationType.BETWEENINCLUSIVERIGHT) {
						found = fsNew.doubleValue == fsOri.doubleValue
								&& fsNew.doubleValue2 == fsOri.doubleValue2;
					}
				}
			}
			if (!found)
				ori.add(newResults.get(i));
		}

		return ori;
	}

	// for each field of a stream -> remove from Hash if wildcard available
	// -> order FilterStatements by the size of their ranges
	private HashMap<Long, HashMap<String, Vector<FilterStatement>>> simplifyAndOrderResultingHashMap(
			HashMap<Long, HashMap<String, Vector<FilterStatement>>> ori) {
		HashMap<Long, HashMap<String, Vector<FilterStatement>>> result = new HashMap<Long, HashMap<String, Vector<FilterStatement>>>();

		Iterator<Long> iteratorID = ori.keySet().iterator();
		while (iteratorID.hasNext()) {
			Long ID = (Long) iteratorID.next(); // ID
			HashMap<String, Vector<FilterStatement>> prevStatementsByFieldHash = ori
					.get(ID);

			HashMap<String, Vector<FilterStatement>> resultStatementsByFieldHash = new HashMap<String, Vector<FilterStatement>>();
			result.put(ID, resultStatementsByFieldHash);

			Iterator<String> iteratorField = prevStatementsByFieldHash.keySet()
					.iterator();
			while (iteratorField.hasNext()) {
				String field = (String) iteratorField.next();
				Vector<FilterStatement> prevStatementsByField = prevStatementsByFieldHash
						.get(field);
				Vector<FilterStatement> resultStatementsByField = new Vector<FilterStatement>();

				if (checkIfAnyIsWildcard(prevStatementsByField))
					; // don't put it in resultStatementsByFieldHash
				else {
					for (int i = 0; i < prevStatementsByField.size(); ++i) {
						FilterStatement statToPut = prevStatementsByField
								.get(i);
						int indexToPut = 0;

						// one possible value only -> put it at the back
						if (statToPut.relationType == FieldRelationType.EQUALS)
							indexToPut = resultStatementsByField.size();
						// infinity range -> put it at the front
						else if (statToPut.relationType == FieldRelationType.EQUALSORGREATER
								|| statToPut.relationType == FieldRelationType.EQUALSORLESSER
								|| statToPut.relationType == FieldRelationType.GREATER
								|| statToPut.relationType == FieldRelationType.LESSER
								|| statToPut.relationType == FieldRelationType.NOTEQUALS)
							indexToPut = 0;
						// limited range (left and right limit) -> put it for
						// the first EQUAL-statements
						else {
							boolean stop = false;
							for (indexToPut = 0; indexToPut < resultStatementsByField
									.size() && !stop; ++indexToPut) {
								FilterStatement stat = resultStatementsByField
										.get(indexToPut);
								if (stat.relationType == FieldRelationType.EQUALS)
									stop = true;
							}

							if (stop)
								indexToPut--;
						}

						// put the FilterStatement in the result Vector, in the
						// correct order and converted to the required value
						// type
						resultStatementsByField.add(indexToPut,
								convertToRequiredValueType(statToPut));
					}

					// put the ordered Vector in the HashMap
					resultStatementsByFieldHash.put(field,
							resultStatementsByField);
				}
			}
		}

		return result;
	}

	// only double values are checked, for simplified comparison !!!
	// fsNew, only: LESSER, EQUALSORLESSER, EQUALS, EQUALSORGREATER, GREATER
	// fsOld, also: BETWEENEXCLUSIVE, BETWEENINCLUSIVELEFT,
	// BETWEENINCLUSIVERIGHT, BETWEENINCLUSIVELEFTANDRIGHT
	private Vector<FilterStatement> mergeNewFilterStatementWithCurrentFilterStatement(
			FilterStatement _fsNew, FilterStatement _fsOld, boolean isAND) {
		Vector<FilterStatement> result = new Vector<FilterStatement>();

		FilterStatement fsNew = _fsNew.clone();
		FilterStatement fsOld = _fsOld.clone();
		fsNew.wasInOr = !isAND;
		fsOld.wasInOr = !isAND;

		if (fsNew.relationType == FieldRelationType.NOTEQUALS)
			fsNew.wildCard = true;
		if (fsOld.relationType == FieldRelationType.NOTEQUALS)
			fsOld.wildCard = true;

		if (!isAND && fsNew.wildCard) {
			result.add(fsNew);
			return result;
		} else if (!isAND && fsOld.wildCard) {
			result.add(fsOld);
			return result;
		}

		if (isAND && fsNew.wildCard) {
			result.add(fsOld);
			return result;
		} else if (isAND && fsOld.wildCard) {
			result.add(fsNew);
			return result;
		}

		if (fsNew.valueType == CEPLEventDataFieldType.TEXT
				|| fsOld.valueType == CEPLEventDataFieldType.TEXT) {
			if (fsNew.valueType != CEPLEventDataFieldType.TEXT
					|| fsOld.valueType != CEPLEventDataFieldType.TEXT
					|| fsNew.relationType != FieldRelationType.EQUALS
					|| fsOld.relationType != FieldRelationType.EQUALS)
				;
			else {
				if (isAND) {
					if (fsNew.stringValue
							.compareToIgnoreCase(fsOld.stringValue) == 0)
						result.add(fsNew);
				} else {
					result.add(fsOld);
					if (fsNew.stringValue
							.compareToIgnoreCase(fsOld.stringValue) != 0)
						result.add(fsNew);
				}
			}

			return result;
		}

		double newValue = fsNew.doubleValue;
		double oldValue = fsOld.doubleValue;
		double oldValue2 = fsOld.doubleValue2;

		if (fsNew.relationType == FieldRelationType.EQUALS) {
			switch (fsOld.relationType) {
			case EQUALS:
				if (isAND && newValue == oldValue)
					result.add(fsOld);
				else if (!isAND) {
					result.add(fsOld);
					if (!(newValue == oldValue))
						result.add(fsNew);
				}
				break;
			case EQUALSORGREATER:
				if (isAND && newValue >= oldValue)
					result.add(fsNew);
				else if (!isAND) {
					result.add(fsOld);
					if (!(newValue >= oldValue))
						result.add(fsNew);
				}
				break;
			case EQUALSORLESSER:
				if (isAND && newValue <= oldValue)
					result.add(fsNew);
				else if (!isAND) {
					result.add(fsOld);
					if (!(newValue <= oldValue))
						result.add(fsNew);
				}
				break;
			case GREATER:
				if (isAND && newValue > oldValue)
					result.add(fsNew);
				else if (!isAND) {
					if (newValue == oldValue) {
						fsOld.relationType = FieldRelationType.EQUALSORGREATER;
						result.add(fsOld);
					} else {
						result.add(fsOld);
						if (!(newValue > oldValue))
							result.add(fsNew);
					}
				}
				break;
			case LESSER:
				if (isAND && newValue < oldValue)
					result.add(fsNew);
				else if (!isAND) {
					if (newValue == oldValue) {
						fsOld.relationType = FieldRelationType.EQUALSORLESSER;
						result.add(fsOld);
					} else {
						result.add(fsOld);
						if (!(newValue < oldValue))
							result.add(fsNew);
					}
				}
				break;
			case BETWEENEXCLUSIVE:
				if (isAND && newValue > oldValue && newValue < oldValue2)
					result.add(fsNew);
				else if (!isAND) {
					if (newValue == oldValue) {
						fsOld.relationType = FieldRelationType.BETWEENINCLUSIVELEFT;
						result.add(fsOld);
					} else if (newValue == oldValue2) {
						fsOld.relationType = FieldRelationType.BETWEENINCLUSIVERIGHT;
						result.add(fsOld);
					} else {
						result.add(fsOld);
						if (!(newValue > oldValue && newValue < oldValue2))
							result.add(fsNew);
					}
				}
				break;
			case BETWEENINCLUSIVELEFT:
				if (isAND && newValue >= oldValue && newValue < oldValue2)
					result.add(fsNew);
				else if (!isAND) {
					if (newValue == oldValue2) {
						fsOld.relationType = FieldRelationType.BETWEENINCLUSIVELEFTANDRIGHT;
						result.add(fsOld);
					} else {
						result.add(fsOld);
						if (!(newValue >= oldValue && newValue < oldValue2))
							result.add(fsNew);
					}
				}
				break;
			case BETWEENINCLUSIVERIGHT:
				if (isAND && newValue > oldValue && newValue <= oldValue2)
					result.add(fsNew);
				else if (!isAND) {
					if (newValue == oldValue) {
						fsOld.relationType = FieldRelationType.BETWEENINCLUSIVELEFTANDRIGHT;
						result.add(fsOld);
					} else {
						result.add(fsOld);
						if (!(newValue > oldValue && newValue <= oldValue2))
							result.add(fsNew);
					}
				}
				break;
			case BETWEENINCLUSIVELEFTANDRIGHT:
				if (isAND && newValue >= oldValue && newValue <= oldValue2)
					result.add(fsNew);
				else if (!isAND) {
					result.add(fsOld);
					if (!(newValue >= oldValue && newValue <= oldValue2))
						result.add(fsNew);
				}
				break;
			}
		} else if (fsNew.relationType == FieldRelationType.EQUALSORGREATER) {
			switch (fsOld.relationType) {
			case EQUALS:
				if (isAND && newValue <= oldValue)
					result.add(fsOld);
				else if (!isAND) {
					result.add(fsNew);
					if (!(newValue <= oldValue))
						result.add(fsOld);
				}
				break;
			case EQUALSORGREATER:
				if (isAND) {
					double max = (newValue > oldValue) ? newValue : oldValue;
					fsNew.doubleValue = max;
					result.add(fsNew);
				} else if (!isAND) {
					if (newValue >= oldValue)
						result.add(fsOld);
					else
						result.add(fsNew);
				}
				break;
			case EQUALSORLESSER:
				if (isAND) {
					if (newValue == oldValue) {
						fsOld.relationType = FieldRelationType.EQUALS;
						result.add(fsOld);
					} else if (newValue < oldValue) {
						fsOld.relationType = FieldRelationType.BETWEENINCLUSIVELEFTANDRIGHT;
						fsOld.doubleValue = newValue;
						fsOld.doubleValue2 = oldValue;
						result.add(fsOld);
					}
				} else if (!isAND) {
					if (newValue <= oldValue) {
						fsOld.wildCard = true; // ANY value is OK
						result.add(fsOld);
					} else {
						result.add(fsOld);
						result.add(fsNew);
					}
				}
				break;
			case GREATER:
				if (isAND) {
					if (oldValue < newValue)
						result.add(fsNew);
					else
						result.add(fsOld);
				} else if (!isAND) {
					if (oldValue >= newValue)
						result.add(fsNew);
					else
						result.add(fsOld);
				}
				break;
			case LESSER:
				if (isAND) {
					if (newValue < oldValue) {
						fsOld.relationType = FieldRelationType.BETWEENINCLUSIVELEFT;
						fsOld.doubleValue = newValue;
						fsOld.doubleValue2 = oldValue;
						result.add(fsOld);
					}
				} else if (!isAND) {
					if (newValue <= oldValue) {
						fsOld.wildCard = true; // ANY value is OK
						result.add(fsOld);
					} else {
						result.add(fsOld);
						result.add(fsNew);
					}
				}
				break;
			case BETWEENEXCLUSIVE:
				if (isAND && newValue < oldValue2) {
					double max = (newValue > oldValue) ? newValue : oldValue;
					if (newValue > oldValue)
						fsOld.relationType = FieldRelationType.BETWEENINCLUSIVELEFT;
					fsOld.doubleValue = max;
					result.add(fsOld);
				} else if (!isAND) {
					if (newValue <= oldValue) {
						result.add(fsNew);
					} else if (newValue > oldValue2) {
						result.add(fsOld);
						result.add(fsNew);
					} else if (newValue <= oldValue2) {
						fsOld.relationType = FieldRelationType.GREATER;
						result.add(fsOld);
					}
				}
				break;
			case BETWEENINCLUSIVELEFT:
				if (isAND && newValue < oldValue2) {
					double max = (newValue > oldValue) ? newValue : oldValue;
					fsOld.doubleValue = max;
					result.add(fsOld);
				} else if (!isAND) {
					if (newValue <= oldValue) {
						result.add(fsNew);
					} else if (newValue > oldValue2) {
						result.add(fsOld);
						result.add(fsNew);
					} else if (newValue <= oldValue2) {
						fsOld.relationType = FieldRelationType.EQUALSORGREATER;
						result.add(fsOld);
					}
				}
				break;
			case BETWEENINCLUSIVERIGHT:
				if (isAND && newValue <= oldValue2) {
					if (newValue == oldValue2) {
						fsNew.relationType = FieldRelationType.EQUALS;
						result.add(fsNew);
					} else {
						double max = (newValue > oldValue) ? newValue
								: oldValue;
						if (max > fsOld.doubleValue)
							fsOld.relationType = FieldRelationType.BETWEENINCLUSIVELEFTANDRIGHT;
						fsOld.doubleValue = max;
						result.add(fsOld);
					}
				} else if (!isAND) {
					if (newValue <= oldValue) {
						result.add(fsNew);
					} else if (newValue <= oldValue2) {
						fsOld.relationType = FieldRelationType.GREATER;
						result.add(fsOld);
					} else {
						result.add(fsOld);
						result.add(fsNew);
					}
				}
				break;
			case BETWEENINCLUSIVELEFTANDRIGHT:
				if (isAND && newValue <= oldValue2) {
					if (newValue == oldValue2) {
						fsNew.relationType = FieldRelationType.EQUALS;
						result.add(fsNew);
					} else {
						double max = (newValue > oldValue) ? newValue
								: oldValue;
						fsOld.doubleValue = max;
						result.add(fsOld);
					}
				} else if (!isAND) {
					if (newValue <= oldValue) {
						result.add(fsNew);
					} else if (newValue <= oldValue2) {
						fsOld.relationType = FieldRelationType.EQUALSORGREATER;
						result.add(fsOld);
					} else {
						result.add(fsOld);
						result.add(fsNew);
					}
				}
				break;
			}
		} else if (fsNew.relationType == FieldRelationType.EQUALSORLESSER) {
			switch (fsOld.relationType) {
			case EQUALS:
				if (isAND && newValue >= oldValue)
					result.add(fsOld);
				else if (!isAND) {
					result.add(fsNew);
					if (!(newValue >= oldValue))
						result.add(fsOld);
				}
				break;
			case EQUALSORGREATER:
				if (isAND) {
					if (newValue == oldValue) {
						fsNew.relationType = FieldRelationType.EQUALS;
						result.add(fsNew);
					} else if (oldValue < newValue) {
						fsOld.relationType = FieldRelationType.BETWEENINCLUSIVELEFTANDRIGHT;
						fsOld.doubleValue2 = newValue;
						result.add(fsOld);
					}
				} else if (!isAND) {
					if (newValue >= oldValue) {
						fsOld.wildCard = true; // ANY value is OK
						result.add(fsOld);
					} else {
						result.add(fsOld);
						result.add(fsNew);
					}
				}
				break;
			case EQUALSORLESSER:
				if (isAND) {
					if (oldValue > newValue)
						result.add(fsNew);
					else
						result.add(fsOld);
				} else if (!isAND) {
					if (newValue <= oldValue)
						result.add(fsOld);
					else
						result.add(fsNew);
				}
				break;
			case GREATER:
				if (isAND) {
					if (oldValue < newValue) {
						fsOld.relationType = FieldRelationType.BETWEENINCLUSIVERIGHT;
						fsOld.doubleValue2 = newValue;
						result.add(fsOld);
					}
				} else if (!isAND) {
					if (oldValue <= newValue) {
						fsOld.wildCard = true; // ANY value is OK
						result.add(fsOld);
					} else {
						result.add(fsOld);
						result.add(fsNew);
					}
				}
				break;
			case LESSER:
				if (isAND) {
					if (oldValue > newValue)
						result.add(fsNew);
					else
						result.add(fsOld);
				} else if (!isAND) {
					if (newValue >= oldValue)
						result.add(fsNew);
					else
						result.add(fsOld);
				}
				break;
			case BETWEENEXCLUSIVE:
				if (isAND && newValue > oldValue) {
					double min = (newValue < oldValue2) ? newValue : oldValue2;
					if (newValue < oldValue2)
						fsOld.relationType = FieldRelationType.BETWEENINCLUSIVERIGHT;
					fsOld.doubleValue2 = min;
					result.add(fsOld);
				} else if (!isAND) {
					if (newValue >= oldValue2) {
						result.add(fsNew);
					} else if (newValue < oldValue) {
						result.add(fsOld);
						result.add(fsNew);
					} else if (newValue >= oldValue) {
						fsOld.relationType = FieldRelationType.LESSER;
						fsOld.doubleValue = oldValue2;
						result.add(fsOld);
					}
				}
				break;
			case BETWEENINCLUSIVELEFT:
				if (isAND && newValue >= oldValue) {
					if (newValue == oldValue) {
						fsNew.relationType = FieldRelationType.EQUALS;
						result.add(fsNew);
					} else {
						double min = (newValue < oldValue2) ? newValue
								: oldValue2;
						if (min < fsOld.doubleValue2)
							fsOld.relationType = FieldRelationType.BETWEENINCLUSIVELEFTANDRIGHT;
						fsOld.doubleValue2 = min;
						result.add(fsOld);
					}
				} else if (!isAND) {
					if (newValue >= oldValue2) {
						result.add(fsNew);
					} else if (newValue >= oldValue) {
						fsOld.relationType = FieldRelationType.LESSER;
						fsOld.doubleValue = oldValue2;
						result.add(fsOld);
					} else {
						result.add(fsOld);
						result.add(fsNew);
					}
				}
				break;
			case BETWEENINCLUSIVERIGHT:
				if (isAND && newValue > oldValue) {
					double min = (newValue < oldValue2) ? newValue : oldValue2;
					fsOld.doubleValue2 = min;
					result.add(fsOld);
				} else if (!isAND) {
					if (newValue >= oldValue2) {
						result.add(fsNew);
					} else if (newValue >= oldValue) {
						fsOld.relationType = FieldRelationType.EQUALSORLESSER;
						fsOld.doubleValue = oldValue2;
						result.add(fsOld);
					} else {
						result.add(fsOld);
						result.add(fsNew);
					}
				}
				break;
			case BETWEENINCLUSIVELEFTANDRIGHT:
				if (isAND && newValue >= oldValue) {
					if (oldValue == newValue) {
						fsNew.relationType = FieldRelationType.EQUALS;
						result.add(fsNew);
					} else {
						double min = (newValue < oldValue2) ? newValue
								: oldValue2;
						fsOld.doubleValue2 = min;
						result.add(fsOld);
					}
				} else if (!isAND) {
					if (newValue >= oldValue2) {
						result.add(fsNew);
					} else if (newValue >= oldValue) {
						fsOld.relationType = FieldRelationType.EQUALSORLESSER;
						fsOld.doubleValue = oldValue2;
						result.add(fsOld);
					} else {
						result.add(fsOld);
						result.add(fsNew);
					}
				}
				break;
			}
		} else if (fsNew.relationType == FieldRelationType.GREATER) {
			switch (fsOld.relationType) {
			case EQUALS:
				if (isAND && newValue < oldValue)
					result.add(fsOld);
				else if (!isAND) {
					if (oldValue == newValue) {
						fsNew.relationType = FieldRelationType.EQUALSORGREATER;
						result.add(fsNew);
					} else {
						result.add(fsNew);
						if (!(newValue < oldValue))
							result.add(fsOld);
					}
				}
				break;
			case EQUALSORGREATER:
				if (isAND) {
					if (newValue >= oldValue)
						result.add(fsNew);
					else
						result.add(fsOld);
				} else if (!isAND) {
					if (newValue >= oldValue)
						result.add(fsOld);
					else
						result.add(fsNew);
				}
				break;
			case EQUALSORLESSER:
				if (isAND) {
					if (newValue < oldValue) {
						fsOld.relationType = FieldRelationType.BETWEENINCLUSIVERIGHT;
						fsOld.doubleValue = newValue;
						fsOld.doubleValue2 = oldValue;
						result.add(fsOld);
					}
				} else if (!isAND) {
					if (newValue <= oldValue) {
						fsOld.wildCard = true; // ANY value is OK
						result.add(fsOld);
					} else {
						result.add(fsOld);
						result.add(fsNew);
					}
				}
				break;
			case GREATER:
				if (isAND) {
					if (oldValue < newValue)
						result.add(fsNew);
					else
						result.add(fsOld);
				} else if (!isAND) {
					if (oldValue >= newValue)
						result.add(fsNew);
					else
						result.add(fsOld);
				}
				break;
			case LESSER:
				if (isAND) {
					if (newValue < oldValue) {
						fsOld.relationType = FieldRelationType.BETWEENEXCLUSIVE;
						fsOld.doubleValue = newValue;
						fsOld.doubleValue2 = oldValue;
						result.add(fsOld);
					}
				} else if (!isAND) {
					if (newValue < oldValue) {
						fsOld.wildCard = true; // ANY value is OK
						result.add(fsOld);
					} else {
						result.add(fsOld);
						result.add(fsNew);
					}
				}
				break;
			case BETWEENEXCLUSIVE:
				if (isAND && newValue < oldValue2) {
					double max = (newValue > oldValue) ? newValue : oldValue;
					fsOld.doubleValue = max;
					result.add(fsOld);
				} else if (!isAND) {
					if (newValue <= oldValue) {
						result.add(fsNew);
					} else if (newValue >= oldValue2) {
						result.add(fsOld);
						result.add(fsNew);
					} else if (newValue < oldValue2) {
						fsOld.relationType = FieldRelationType.GREATER;
						result.add(fsOld);
					}
				}
				break;
			case BETWEENINCLUSIVELEFT:
				if (isAND && newValue < oldValue2) {
					double max = (newValue > oldValue) ? newValue : oldValue;
					if (newValue >= oldValue)
						fsOld.relationType = FieldRelationType.BETWEENEXCLUSIVE;
					fsOld.doubleValue = max;
					result.add(fsOld);
				} else if (!isAND) {
					if (newValue < oldValue) {
						result.add(fsNew);
					} else if (newValue < oldValue2) {
						fsNew.relationType = FieldRelationType.EQUALSORGREATER;
						result.add(fsNew);
					} else if (newValue >= oldValue2) {
						result.add(fsOld);
						result.add(fsNew);
					}
				}
				break;
			case BETWEENINCLUSIVERIGHT:
				if (isAND && newValue < oldValue2) {
					double max = (newValue > oldValue) ? newValue : oldValue;
					fsOld.doubleValue = max;
					result.add(fsOld);
				} else if (!isAND) {
					if (newValue <= oldValue) {
						result.add(fsNew);
					} else if (newValue <= oldValue2) {
						fsOld.relationType = FieldRelationType.GREATER;
						result.add(fsOld);
					} else {
						result.add(fsOld);
						result.add(fsNew);
					}
				}
				break;
			case BETWEENINCLUSIVELEFTANDRIGHT:
				if (isAND && newValue < oldValue2) {
					double max = (newValue > oldValue) ? newValue : oldValue;
					if (newValue >= oldValue)
						fsOld.relationType = FieldRelationType.BETWEENINCLUSIVERIGHT;
					fsOld.doubleValue = max;
					result.add(fsOld);
				} else if (!isAND) {
					if (newValue < oldValue) {
						result.add(fsNew);
					} else if (newValue <= oldValue2) {
						fsOld.relationType = FieldRelationType.EQUALSORGREATER;
						result.add(fsOld);
					} else {
						result.add(fsOld);
						result.add(fsNew);
					}
				}
				break;
			}
		} else if (fsNew.relationType == FieldRelationType.LESSER) {
			switch (fsOld.relationType) {
			case EQUALS:
				if (isAND && newValue > oldValue)
					result.add(fsOld);
				else if (!isAND) {
					if (oldValue == newValue) {
						fsNew.relationType = FieldRelationType.EQUALSORLESSER;
						result.add(fsNew);
					} else {
						result.add(fsNew);
						if (!(newValue > oldValue))
							result.add(fsOld);
					}
				}
				break;
			case EQUALSORGREATER:
				if (isAND) {
					if (oldValue < newValue) {
						fsOld.relationType = FieldRelationType.BETWEENINCLUSIVELEFT;
						fsOld.doubleValue2 = newValue;
						result.add(fsOld);
					}
				} else if (!isAND) {
					if (newValue >= oldValue) {
						fsOld.wildCard = true; // ANY value is OK
						result.add(fsOld);
					} else {
						result.add(fsOld);
						result.add(fsNew);
					}
				}
				break;
			case EQUALSORLESSER:
				if (isAND) {
					if (oldValue >= newValue)
						result.add(fsNew);
					else
						result.add(fsOld);
				} else if (!isAND) {
					if (newValue <= oldValue)
						result.add(fsOld);
					else
						result.add(fsNew);
				}
				break;
			case GREATER:
				if (isAND) {
					if (oldValue < newValue) {
						fsOld.relationType = FieldRelationType.BETWEENEXCLUSIVE;
						fsOld.doubleValue2 = newValue;
						result.add(fsOld);
					}
				} else if (!isAND) {
					if (oldValue < newValue) {
						fsOld.wildCard = true; // ANY value is OK
						result.add(fsOld);
					} else {
						result.add(fsOld);
						result.add(fsNew);
					}
				}
				break;
			case LESSER:
				if (isAND) {
					if (oldValue > newValue)
						result.add(fsNew);
					else
						result.add(fsOld);
				} else if (!isAND) {
					if (newValue >= oldValue)
						result.add(fsNew);
					else
						result.add(fsOld);
				}
				break;
			case BETWEENEXCLUSIVE:
				if (isAND && newValue > oldValue) {
					double min = (newValue < oldValue2) ? newValue : oldValue2;
					fsOld.doubleValue2 = min;
					result.add(fsOld);
				} else if (!isAND) {
					if (newValue >= oldValue2) {
						result.add(fsNew);
					} else if (newValue < oldValue) {
						result.add(fsOld);
						result.add(fsNew);
					} else if (newValue >= oldValue) {
						fsOld.relationType = FieldRelationType.LESSER;
						fsOld.doubleValue = oldValue2;
						result.add(fsOld);
					}
				}
				break;
			case BETWEENINCLUSIVELEFT:
				if (isAND && newValue > oldValue) {
					double min = (newValue < oldValue2) ? newValue : oldValue2;
					fsOld.doubleValue2 = min;
					result.add(fsOld);
				} else if (!isAND) {
					if (newValue > oldValue2) {
						result.add(fsNew);
					} else if (newValue >= oldValue) {
						fsOld.relationType = FieldRelationType.LESSER;
						fsOld.doubleValue = oldValue2;
						result.add(fsOld);
					} else {
						result.add(fsOld);
						result.add(fsNew);
					}
				}
				break;
			case BETWEENINCLUSIVERIGHT:
				if (isAND && newValue > oldValue) {
					double min = (newValue < oldValue2) ? newValue : oldValue2;
					if (newValue <= oldValue2)
						fsOld.relationType = FieldRelationType.BETWEENEXCLUSIVE;
					fsOld.doubleValue2 = min;
					result.add(fsOld);
				} else if (!isAND) {
					if (newValue > oldValue2) {
						result.add(fsNew);
					} else if (newValue >= oldValue) {
						fsOld.relationType = FieldRelationType.EQUALSORLESSER;
						fsOld.doubleValue = oldValue2;
						result.add(fsOld);
					} else {
						result.add(fsOld);
						result.add(fsNew);
					}
				}
				break;
			case BETWEENINCLUSIVELEFTANDRIGHT:
				if (isAND && newValue > oldValue) {
					double min = (newValue < oldValue2) ? newValue : oldValue2;
					if (newValue <= oldValue2)
						fsOld.relationType = FieldRelationType.BETWEENINCLUSIVELEFT;
					fsOld.doubleValue2 = min;
					result.add(fsOld);
				} else if (!isAND) {
					if (newValue > oldValue2) {
						result.add(fsNew);
					} else if (newValue >= oldValue) {
						fsOld.relationType = FieldRelationType.EQUALSORLESSER;
						fsOld.doubleValue = oldValue2;
						result.add(fsOld);
					} else {
						result.add(fsOld);
						result.add(fsNew);
					}
				}
				break;
			}
		}

		return result;
	}

	public synchronized void processEvent(Event e) {
		HashMap<String, Vector<FilterStatement>> filters = streamIDToFilterStatements
				.get(e.getStreamID());

		// never used in a query
		if (filters == null)
			return;
		else if (mayPass(e, filters))
			next.processEvent(e);
	}

	private boolean mayPass(Event e,
			HashMap<String, Vector<FilterStatement>> filters) {
		if (filters.size() == 0) // ALL events of this type may pass
			return true;
		else {
			// for each field in filters min. one valid FilterStatement -> pass
			// field with wildcard -> not present in <filters>
			boolean pass = true;

			Iterator<String> itField = filters.keySet().iterator();
			while (itField.hasNext() && pass) {
				String field = itField.next();
				Vector<FilterStatement> filter = filters.get(field);

				boolean passForField = false;

				for (int i = 0; i < filter.size() && !passForField; ++i) {
					FilterStatement fs = filter.get(i);
					try {
						if (fs.valueType == CEPLEventDataFieldType.CHAR) {
							char val = e.getCharacter(fs.field);

							if (fs.relationType == FieldRelationType.EQUALSORGREATER
									&& val >= fs.charValue)
								passForField = true;
							else if (fs.relationType == FieldRelationType.EQUALSORLESSER
									&& val <= fs.charValue)
								passForField = true;
							else if (fs.relationType == FieldRelationType.GREATER
									&& val > fs.charValue)
								passForField = true;
							else if (fs.relationType == FieldRelationType.LESSER
									&& val < fs.charValue)
								passForField = true;
							else if (fs.relationType == FieldRelationType.BETWEENINCLUSIVELEFTANDRIGHT
									&& val >= fs.charValue
									&& val <= fs.charValue2)
								passForField = true;
							else if (fs.relationType == FieldRelationType.BETWEENINCLUSIVELEFT
									&& val >= fs.charValue
									&& val < fs.charValue2)
								passForField = true;
							else if (fs.relationType == FieldRelationType.BETWEENINCLUSIVERIGHT
									&& val > fs.charValue
									&& val <= fs.charValue2)
								passForField = true;
							else if (fs.relationType == FieldRelationType.BETWEENEXCLUSIVE
									&& val > fs.charValue
									&& val < fs.charValue2)
								passForField = true;
							else if (fs.relationType == FieldRelationType.EQUALS
									&& val == fs.charValue)
								passForField = true;
						} else if (fs.valueType == CEPLEventDataFieldType.SHORT) {
							short val = e.getShort(fs.field);

							if (fs.relationType == FieldRelationType.EQUALSORGREATER
									&& val >= fs.shortValue)
								passForField = true;
							else if (fs.relationType == FieldRelationType.EQUALSORLESSER
									&& val <= fs.shortValue)
								passForField = true;
							else if (fs.relationType == FieldRelationType.GREATER
									&& val > fs.shortValue)
								passForField = true;
							else if (fs.relationType == FieldRelationType.LESSER
									&& val < fs.shortValue)
								passForField = true;
							else if (fs.relationType == FieldRelationType.BETWEENINCLUSIVELEFTANDRIGHT
									&& val >= fs.shortValue
									&& val <= fs.shortValue2)
								passForField = true;
							else if (fs.relationType == FieldRelationType.BETWEENINCLUSIVELEFT
									&& val >= fs.shortValue
									&& val < fs.shortValue2)
								passForField = true;
							else if (fs.relationType == FieldRelationType.BETWEENINCLUSIVERIGHT
									&& val > fs.shortValue
									&& val <= fs.shortValue2)
								passForField = true;
							else if (fs.relationType == FieldRelationType.BETWEENEXCLUSIVE
									&& val > fs.shortValue
									&& val < fs.shortValue2)
								passForField = true;
							else if (fs.relationType == FieldRelationType.EQUALS
									&& val == fs.shortValue)
								passForField = true;
						} else if (fs.valueType == CEPLEventDataFieldType.INT) {
							int val = e.getInteger(fs.field);

							if (fs.relationType == FieldRelationType.EQUALSORGREATER
									&& val >= fs.intValue)
								passForField = true;
							else if (fs.relationType == FieldRelationType.EQUALSORLESSER
									&& val <= fs.intValue)
								passForField = true;
							else if (fs.relationType == FieldRelationType.GREATER
									&& val > fs.intValue)
								passForField = true;
							else if (fs.relationType == FieldRelationType.LESSER
									&& val < fs.intValue)
								passForField = true;
							else if (fs.relationType == FieldRelationType.BETWEENINCLUSIVELEFTANDRIGHT
									&& val >= fs.intValue
									&& val <= fs.intValue2)
								passForField = true;
							else if (fs.relationType == FieldRelationType.BETWEENINCLUSIVELEFT
									&& val >= fs.intValue && val < fs.intValue2)
								passForField = true;
							else if (fs.relationType == FieldRelationType.BETWEENINCLUSIVERIGHT
									&& val > fs.intValue && val <= fs.intValue2)
								passForField = true;
							else if (fs.relationType == FieldRelationType.BETWEENEXCLUSIVE
									&& val > fs.intValue && val < fs.intValue2)
								passForField = true;
							else if (fs.relationType == FieldRelationType.EQUALS
									&& val == fs.intValue)
								passForField = true;
						} else if (fs.valueType == CEPLEventDataFieldType.LONG) {
							long val = e.getLong(fs.field);

							if (fs.relationType == FieldRelationType.EQUALSORGREATER
									&& val >= fs.longValue)
								passForField = true;
							else if (fs.relationType == FieldRelationType.EQUALSORLESSER
									&& val <= fs.longValue)
								passForField = true;
							else if (fs.relationType == FieldRelationType.GREATER
									&& val > fs.longValue)
								passForField = true;
							else if (fs.relationType == FieldRelationType.LESSER
									&& val < fs.longValue)
								passForField = true;
							else if (fs.relationType == FieldRelationType.BETWEENINCLUSIVELEFTANDRIGHT
									&& val >= fs.longValue
									&& val <= fs.longValue2)
								passForField = true;
							else if (fs.relationType == FieldRelationType.BETWEENINCLUSIVELEFT
									&& val >= fs.longValue
									&& val < fs.longValue2)
								passForField = true;
							else if (fs.relationType == FieldRelationType.BETWEENINCLUSIVERIGHT
									&& val > fs.longValue
									&& val <= fs.longValue2)
								passForField = true;
							else if (fs.relationType == FieldRelationType.BETWEENEXCLUSIVE
									&& val > fs.longValue
									&& val < fs.longValue2)
								passForField = true;
							else if (fs.relationType == FieldRelationType.EQUALS
									&& val == fs.longValue)
								passForField = true;
						} else if (fs.valueType == CEPLEventDataFieldType.FLOAT) {
							float val = e.getFloat(fs.field);

							if (fs.relationType == FieldRelationType.EQUALSORGREATER
									&& val >= fs.floatValue)
								passForField = true;
							else if (fs.relationType == FieldRelationType.EQUALSORLESSER
									&& val <= fs.floatValue)
								passForField = true;
							else if (fs.relationType == FieldRelationType.GREATER
									&& val > fs.floatValue)
								passForField = true;
							else if (fs.relationType == FieldRelationType.LESSER
									&& val < fs.floatValue)
								passForField = true;
							else if (fs.relationType == FieldRelationType.BETWEENINCLUSIVELEFTANDRIGHT
									&& val >= fs.floatValue
									&& val <= fs.floatValue2)
								passForField = true;
							else if (fs.relationType == FieldRelationType.BETWEENINCLUSIVELEFT
									&& val >= fs.floatValue
									&& val < fs.floatValue2)
								passForField = true;
							else if (fs.relationType == FieldRelationType.BETWEENINCLUSIVERIGHT
									&& val > fs.floatValue
									&& val <= fs.floatValue2)
								passForField = true;
							else if (fs.relationType == FieldRelationType.BETWEENEXCLUSIVE
									&& val > fs.floatValue
									&& val < fs.floatValue2)
								passForField = true;
							else if (fs.relationType == FieldRelationType.EQUALS
									&& val == fs.floatValue)
								passForField = true;
						} else if (fs.valueType == CEPLEventDataFieldType.DOUBLE) {
							double val = e.getDouble(fs.field);

							if (fs.relationType == FieldRelationType.EQUALSORGREATER
									&& val >= fs.doubleValue)
								passForField = true;
							else if (fs.relationType == FieldRelationType.EQUALSORLESSER
									&& val <= fs.doubleValue)
								passForField = true;
							else if (fs.relationType == FieldRelationType.GREATER
									&& val > fs.doubleValue)
								passForField = true;
							else if (fs.relationType == FieldRelationType.LESSER
									&& val < fs.doubleValue)
								passForField = true;
							else if (fs.relationType == FieldRelationType.BETWEENINCLUSIVELEFTANDRIGHT
									&& val >= fs.doubleValue
									&& val <= fs.doubleValue2)
								passForField = true;
							else if (fs.relationType == FieldRelationType.BETWEENINCLUSIVELEFT
									&& val >= fs.doubleValue
									&& val < fs.doubleValue2)
								passForField = true;
							else if (fs.relationType == FieldRelationType.BETWEENINCLUSIVERIGHT
									&& val > fs.doubleValue
									&& val <= fs.doubleValue2)
								passForField = true;
							else if (fs.relationType == FieldRelationType.BETWEENEXCLUSIVE
									&& val > fs.doubleValue
									&& val < fs.doubleValue2)
								passForField = true;
							else if (fs.relationType == FieldRelationType.EQUALS
									&& val == fs.doubleValue)
								passForField = true;
						} else if (fs.valueType == CEPLEventDataFieldType.TEXT) {
							String val = e.getString(fs.field);
							if (fs.relationType == FieldRelationType.EQUALS
									&& val.compareToIgnoreCase(fs.stringValue) == 0)
								passForField = true;
						}
					} catch (UnknownDataFieldException e1) {
						e1.printStackTrace();
						return false;
					}
				}

				if (passForField == false)
					pass = false;
			}

			return pass;
		}
	}
}
